package com.atguigu.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

//首页和详情页的访问数据
public class DwdTrafficHomeAndGoodDetail {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(3 * 60 * 1000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */

        // TODO 3 读取kafka中dwd_traffic_page_log主题数据
        String topicName = "dwd_traffic_page_log";
        String groupId = "dwd_traffic_home_and_good_detail";
        DataStreamSource<String> pageStream = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topicName, groupId));

        // TODO 4 过滤数据并转换格式
        SingleOutputStreamOperator<JSONObject> jsonObjStream = pageStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String pageId = jsonObject.getJSONObject("page").getString("page_id");
                if ("home".equals(pageId) || "good_detail".equals(pageId)){
                    collector.collect(jsonObject);
                }
            }
        });

        // TODO 5 独立访客去重
        KeyedStream<JSONObject, String> keyedStream = jsonObjStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("mid");
            }
        });

        SingleOutputStreamOperator<JSONObject> processStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            ValueState<String> homeLastVisitDtState = null;
            ValueState<String> detailLastVisitDtState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> homeLastVisitDtDescriptor = new ValueStateDescriptor<>("home_last_visit_dt", String.class);
                homeLastVisitDtDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build());
                homeLastVisitDtState = getRuntimeContext().getState(homeLastVisitDtDescriptor);
                ValueStateDescriptor<String> detailLastVisitDtDescriptor = new ValueStateDescriptor<>("detail_last_visit_dt", String.class);
                detailLastVisitDtDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build());
                detailLastVisitDtState = getRuntimeContext().getState(detailLastVisitDtDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObject, Context context, Collector<JSONObject> collector) throws Exception {
                String homeLastVisitDt = homeLastVisitDtState.value();
                String detailLasiVisitDt = detailLastVisitDtState.value();
                String pageId = jsonObject.getJSONObject("page").getString("page_id");
                Long ts = jsonObject.getLong("ts");
                String visitDt = DateFormatUtil.toDate(ts);
                if ("home".equals(pageId)) {
                    if (homeLastVisitDt == null || !homeLastVisitDt.equals(visitDt)) {
                        collector.collect(jsonObject);
                        homeLastVisitDtState.update(visitDt);
                    }
                } else {
                    if (detailLasiVisitDt == null || !detailLasiVisitDt.equals(visitDt)) {
                        collector.collect(jsonObject);
                        detailLastVisitDtState.update(visitDt);
                    }
                }
            }
        });
        processStream.print("process>>>");

        // TODO 6 写出到kafka中dwd_traffic_home_and_good_detail主题
        String homeAndGoodDetailTopic = "dwd_traffic_home_and_good_detail";
        processStream.map(new MapFunction<JSONObject, String>() {
            @Override
            public String map(JSONObject jsonObject) throws Exception {
                return jsonObject.toJSONString();
            }
        })
                .addSink(KafkaUtil.getFlinkKafkaProducer(homeAndGoodDetailTopic));

        // TODO 7 执行任务
        env.execute(groupId);
    }
}
